package com.wangbloc.domain.service.impl;

import cn.hutool.core.text.StrPool;
import cn.hutool.core.util.IdUtil;
import com.wangbloc.common.MqttProperties;
import com.wangbloc.domain.callback.CustomMqttCallback;
import com.wangbloc.domain.service.MqttService;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static org.eclipse.paho.client.mqttv3.MqttException.REASON_CODE_CLIENT_EXCEPTION;

/**
 * @author wangbloc
 * @version 1.0
 * @date 2023/2/9 22:39
 **/
@Service
public class MqttServiceImpl implements MqttService {

    @Resource
    private MqttProperties mqttProperties;

    private static MqttClient MQTT_CLIENT_DEFAULT;

    private static String MQTT_CLIENT_ID_UUID = StrPool.UNDERLINE + IdUtil.getSnowflakeNextIdStr();

    public static Set<String> SUB_TOPIC = new HashSet<>();

    private ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(10);

    @Override
    public MqttClient getDefaultClient() {
        // 如果本地缓存中有连接，且连接信息有效，则直接返回
        if(null == MQTT_CLIENT_DEFAULT || !MQTT_CLIENT_DEFAULT.isConnected()){
            try {
                // 如果本地缓存连接无效，则调用关闭方法，且将缓存重置为null
                if (null != MQTT_CLIENT_DEFAULT) {
                    MQTT_CLIENT_DEFAULT.close();
                    MQTT_CLIENT_DEFAULT = null;
                }
                MemoryPersistence persistence = new MemoryPersistence();
                MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
                mqttConnectOptions.setCleanSession(true);
                mqttConnectOptions.setUserName(mqttProperties.getUserName());
                mqttConnectOptions.setPassword(mqttProperties.getPassword().toCharArray());
                mqttConnectOptions.setConnectionTimeout(30);
                mqttConnectOptions.setKeepAliveInterval(10);
                mqttConnectOptions.setServerURIs(mqttProperties.getBroker().split(StrPool.COMMA));
                MQTT_CLIENT_DEFAULT = new MqttClient(mqttProperties.getBroker(),
                        mqttProperties.getClientId() + MQTT_CLIENT_ID_UUID, persistence);
                MQTT_CLIENT_DEFAULT.setCallback(new CustomMqttCallback());
                MQTT_CLIENT_DEFAULT.connect(mqttConnectOptions);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        return MQTT_CLIENT_DEFAULT;
    }

    @Override
    public void customSubscribe(String topic) {
        if(null == MQTT_CLIENT_DEFAULT || !MQTT_CLIENT_DEFAULT.isConnected()){
            getDefaultClient();
        }
        try {
            MQTT_CLIENT_DEFAULT.subscribe(topic);
        }catch (Exception e){
            e.printStackTrace();
        }
        SUB_TOPIC.add(topic);
    }

    @Override
    public void lostAutoSubscribe() {
        if(SUB_TOPIC.isEmpty()){
            return;
        }
        try {
            MqttClient mqttClient = getDefaultClient();
            if (null == mqttClient || !mqttClient.isConnected()) {
                throw new MqttException(REASON_CODE_CLIENT_EXCEPTION);
            }
            for (String topic : SUB_TOPIC) {
                mqttClient.subscribe(topic);
            }
        }catch (MqttException e){
            e.printStackTrace();
            SCHEDULED_EXECUTOR_SERVICE.schedule(()->{
                lostAutoSubscribe();
            }, 1, TimeUnit.MINUTES);
        }catch (Exception e){
            SCHEDULED_EXECUTOR_SERVICE.schedule(()->{
                lostAutoSubscribe();
            }, 1, TimeUnit.MINUTES);
        }
    }
}
